package com.jiazhong.example.example.rocket.demo1.logistics.consumer;

import com.alibaba.fastjson.JSONArray;
import com.jiazhong.example.example.rocket.demo1.commons.bean.Alipay;
import com.jiazhong.example.example.rocket.demo1.commons.bean.Logistics;
import com.jiazhong.example.example.rocket.demo1.logistics.service.LogisticsService;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
@Component
public class ChangeLogisticsConsumer {
    @Resource
    private LogisticsService logisticsService;

    @Scheduled(initialDelay = 1000)
    @SneakyThrows
    public void saveLogistics() {
        log.info("物流消费者启动成功");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_consumer_group");
        consumer.setNamesrvAddr("124.220.170.254:9876");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.subscribe("Order", "alipay");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                log.info("物流消费者获取到消息");
                for (MessageExt messageExt : list) {
                    byte[] body = messageExt.getBody();
                    String json = new String(body, StandardCharsets.UTF_8);
                    Alipay alipay = JSONArray.parseObject(json, Alipay.class);
                    Logistics logistics = new Logistics();
                    logistics.setAddress(alipay.getAddress());
                    logistics.setName(alipay.getName());
                    logistics.setPhone(alipay.getPhone());
                    logisticsService.save(logistics);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}
